package com.hins.mq.transactionMq;

import com.hins.mq.common.Config;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.*;

/**
 * 本地事务的执行只发生在生产者，因为消费者消费失败会有重试的机制，而事务消息对于生产者来说，会先发送一个half消息，然后broker响应回来信息
 * 说明我们都可以正常交互的意思，然后生产者真正发送信息，且最终得到响应发送成功，这是才会进行commit或者rollback
 * 要是中途本地事务执行时间过长，broker会进行向生产者进行回查机制和补偿机制
 */
public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {

        //创建消息生产者
        TransactionMQProducer producer = new TransactionMQProducer("transaction_group");
        producer.setNamesrvAddr(Config.namesrvAddr);

        //创建事务监听器
        TransactionListener transactionListener = new TransactionListenerImpl();
        //生产者这是监听器，用于执行本地事务和消息回查
        producer.setTransactionListener(transactionListener);

        //线程池
        ExecutorService executorService = new ThreadPoolExecutor(
                2, 5, 100, TimeUnit.SECONDS,
                new ArrayBlockingQueue<Runnable>(2000),
                new ThreadFactory() {
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("client-transaction-msg-check-thread");
                        return thread;
            }
        });

        producer.setExecutorService(executorService);

        //启动消息生产者
        producer.start();
        String[] tags = new String[]{"TagA", "TagB", "TagC"};
        for (int i = 0; i < 1; i++) {
            try {
                Message msg = new Message("TransactionTopic", tags[i % tags.length], "KEY" + i,
                        ("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
                SendResult sendResult = producer.sendMessageInTransaction(msg, null);
                System.out.printf("%s%n", sendResult);
                TimeUnit.SECONDS.sleep(1);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        //producer.shutdown();
    }
}